package com.xxk.rocket.star.core.builder;

import com.alibaba.rocketmq.client.consumer.MQConsumer;
import com.alibaba.rocketmq.client.producer.MQProducer;
import com.xxk.rocket.star.manage.StargateInstance;
import com.xxk.rocket.star.core.annotation.IProcessAnnotationConsume;
import com.xxk.rocket.star.core.annotation.IProcessAnnotationProducer;
import com.xxk.rocket.star.core.client.IProcessClientConsume;
import com.xxk.rocket.star.core.client.IProcessClientProducer;
import com.xxk.rocket.star.core.consume.ConsumeFactory;
import com.xxk.rocket.star.core.consume.ConsumeHandlerImpl;
import com.xxk.rocket.star.core.consume.MessageListener;
import com.xxk.rocket.star.core.context.ConsumeContext;
import com.xxk.rocket.star.core.context.ProducerContext;
import com.xxk.rocket.star.core.producer.ProducerFactory;
import com.xxk.rocket.star.core.producer.ProducerInvocationHandlerImpl;
import com.xxk.rocket.star.manage.StargateConsumeManageImpl;
import com.xxk.rocket.star.manage.StargateProcessManageImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

@SuppressWarnings({"unchecked"})
public class StargateFactory {

  private static final Logger log = LoggerFactory.getLogger(StargateFactory.class);

  private static void init() {
    if (!StargateInstance.getStargateInstance().isInit()) {
      StargateInstance.getStargateInstance()
              .init(new StargateProcessManageImpl(), new StargateConsumeManageImpl());
    }
  }

  public static <T> T createProducer(Class<T> clazz, ProducerContext producerContextInit)
          throws Throwable {
    init();
    //生成实例
    log.debug("new ProducerInvocationHandlerImpl for Class {}", clazz.getName());
    ProducerInvocationHandlerImpl handler = new ProducerInvocationHandlerImpl();
    T o = (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, handler);

    log.debug("clone ProducerContext for class {}", clazz.getName());
    ProducerContext producerContext = producerContextInit.clone();
    for (IProcessAnnotationProducer processProducer : ProcessCenter
            .getProducersAnnotationProcess()) {
      log.debug(
              "IProcessAnnotationProducer for Class {} ing", processProducer.getClass().getName());
      processProducer.processProducerByContext(producerContext, clazz, o, null);
      if (!producerContext.isValid()) {
        log.info("class is not Valid {}", clazz.getName());
        return null;
      }
    }

    Method[] methods = clazz.getMethods();
    for (Method method : methods) {
      log.debug("clone ProducerContext for method {}", method.getName());
      ProducerContext producerContextMethod = producerContext.clone();
      for (IProcessAnnotationProducer processProducer : ProcessCenter
              .getProducersAnnotationProcess()) {
        log.debug(
                "IProcessAnnotationProducer for Method {} ing", processProducer.getClass().getName());
        processProducer.processProducerByContext(producerContextMethod, clazz, o, method);
      }
      if (!producerContextMethod.isValid()) {
        log.info("method is not Valid {}", method.getName());
        continue;
      }
      log.debug("create MQProducer {}", method.getName());
      MQProducer mqProducer = ProducerFactory.getProducer(producerContextMethod);
      producerContextMethod.setMqProducer(mqProducer);
      handler.putContextMap(method, producerContextMethod);
      for (IProcessClientProducer processClientProducer : ProcessCenter
              .getProducersClientProcess()) {
        log.debug("IProcessClientProducer for Method {} ing",
                processClientProducer.getClass().getName());
        processClientProducer.processProducerByContext(producerContextMethod, mqProducer);
      }
      log.info("{} MQProducer start", method.getName());
      mqProducer.start();
    }
    return o;
  }

  public static <T> T createConsume(Class<T> clazz, ConsumeContext consumeContextInit)
          throws Throwable {
    init();
    log.debug("new Instance for Class {}", clazz.getName());
    T o = clazz.getDeclaredConstructor().newInstance();

    log.debug("clone ConsumeContext for class {}", clazz.getName());
    ConsumeContext consumeContext = consumeContextInit.clone();
    for (IProcessAnnotationConsume processConsume : ProcessCenter.getConsumeAnnotationProcess()) {
      log.debug(
              "IProcessAnnotationConsume for Class {} ing", processConsume.getClass().getName());
      processConsume.processConsumeByContext(consumeContext, clazz, o, null);
      if (!consumeContext.isValid()) {
        log.info("class is not Valid {}", clazz.getName());
        return null;
      }
    }

    Method[] methods = clazz.getMethods();
    ConsumeHandlerImpl consumeHandler = new ConsumeHandlerImpl(consumeContext.getThreadSize());
    for (Method method : methods) {
      log.debug("clone ConsumeContext for method {}", method.getName());
      ConsumeContext consumeContextMethod = consumeContext.clone();
      for (IProcessAnnotationConsume processConsume : ProcessCenter.getConsumeAnnotationProcess()) {
        log.debug(
                "IProcessAnnotationConsume for method {} ing", processConsume.getClass().getName());
        processConsume.processConsumeByContext(consumeContextMethod, clazz, o, method);
      }
      if (!consumeContextMethod.isValid()) {
        log.info("method is not Valid {}", clazz.getName());
        continue;
      }
      log.debug("create MQConsumer {}", method.getName());
      MQConsumer mqConsumer = ConsumeFactory.getConsume(consumeContextMethod);
      for (IProcessClientConsume processClientConsume : ProcessCenter
              .getConsumeClientProcess()) {
        log.debug(
                "IProcessClientConsume for method {} ing", processClientConsume.getClass().getName());
        processClientConsume.processConsumeByContext(consumeContextMethod, mqConsumer);
      }

      MessageListener messageListener = new MessageListener();
      messageListener.setConsumeContext(consumeContextMethod);
      messageListener.setConsumeHandler(consumeHandler);
      ConsumeFactory.registerMessageListener(mqConsumer, messageListener);

    }
    return o;
  }
}
